package com.gitee.funcy.disruptorlearn.ch02.quickstart;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author fangchengyan
 * @date 2019-03-13 11:46
 */
public class Main {

    public static void main(String[] args){
        /**
         *  1. eventFactory：消息（event）工厂对象
         *  2. ringBufferSize：容器的长度
         *  3. executor：线程池
         *  4. 单生产者还是多生产者
         *  5. 等待策略
         */

        OrderEventFactory orderEventFactory = new OrderEventFactory();
        int ringBufferSize = 1024 * 1024;
        int nThreads = Runtime.getRuntime().availableProcessors();

        ExecutorService executor = new ThreadPoolExecutor(
                nThreads,
                nThreads,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );


        // 1. 实例化disruptor对象：持有RingBuffer、消费者线程池Executor、消费者集合ConsumerRepository等引用
        Disruptor<OrderEvent> disruptor = new Disruptor<>(orderEventFactory, ringBufferSize,
                executor, ProducerType.SINGLE,
                //决定一个消费者将如何等待生产者将event置入disruptor中
                new BlockingWaitStrategy()
        );

        // 2. 添加消费者的监听
        disruptor.handleEventsWith(new OrderEventHandler());

        // 3.启动disruptor
        disruptor.start();

        // 4. 获取实际存储数据的容器：ringBuffer
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();

        OrderEventProducer producer = new OrderEventProducer(ringBuffer);

        ByteBuffer bb = ByteBuffer.allocate(8);

        for(long i = 0; i < 100; i++) {
            bb.putLong(0, i);
            producer.sendData(bb);
        }

        disruptor.shutdown();
        executor.shutdown();
    }

}

















